package com.liuguang.rabbitmq.service.impl;

import com.alibaba.fastjson.JSON;
import com.liuguang.api.rabbitmq.RabbitMqService;
import com.liuguang.api.redis.RedisService;
import com.liuguang.common.constans.CommonConstans;
import com.liuguang.common.pojo.MqData;
import com.liuguang.rabbitmq.pojo.MyCorrelationData;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.dubbo.config.annotation.DubboService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.UUID;

/**
 * @ClassName: RabbitMqServiceImpl
 * @Description：想不用再暴露个dubbo服务，然后调用服务，然后产生消息，这样增加了调用服务的开销，直接再调用端产生消息 但是也有个问题，就是每个调用服务地方都要引依赖，重复，
 * 最终还是暴露服务，并且消息产生后回调不可能在每个调用类上加，这是主要原因
 * @Author: wanghr
 * Date: 2:28 PM 2020/10/14
 * Version: 1.0
 */
@DubboService
@Service
public class RabbitMqServiceImpl implements RabbitMqService {
    private static final Logger logger = LoggerFactory.getLogger(RabbitMqServiceImpl.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @DubboReference
    private RedisService redisService;

    @Override
    public void sendEmail(MqData mqData) {
        //消息关联数据，为后序的消息确保被消费时使用
        CorrelationData corrData = new CorrelationData(UUID.randomUUID().toString());
        logger.info("corrData：{}", corrData.getId());
        rabbitTemplate.convertAndSend(CommonConstans.DIRECT_EXCHANGE_NAME, CommonConstans.EMAIL_ROUTING_KEY, mqData, corrData);
    }

    @Override
    public void sendTestMessage(MqData mqData) {
//        CorrelationData corrData = new CorrelationData(UUID.randomUUID().toString());
//        logger.info("corrData：{}", corrData.getId());
//        rabbitTemplate.convertAndSend(CommonConstans.DIRECT_EXCHANGE_NAME, CommonConstans.TEST_ROUTING_KEY, mqData.getMsg(), corrData);

        //测试使用自定义 CorrelationData 类
        MyCorrelationData data = new MyCorrelationData(UUID.randomUUID().toString(), CommonConstans.DIRECT_EXCHANGE_NAME,
                CommonConstans.TEST_ROUTING_KEY, mqData);
        logger.info("corrData：{}", data);
        try {
            rabbitTemplate.convertAndSend(CommonConstans.DIRECT_EXCHANGE_NAME, CommonConstans.TEST_ROUTING_KEY, mqData, data);
        } catch (AmqpConnectException e) {
            //捕获AmqpConnectException 异常，因为这种异常是ack 过程中connection出现异常才会抛出
            // 保存到redis 中，定时任务后序重发
            logger.info("消息发送失败，id：{}");
            String key = CommonConstans.FAIL_MESSAGE_LIST_KEY;
            String value = JSON.toJSONString(data);
            redisService.saveToList(key, value);
        }

    }

    /**
     * 测试消息发送到broker，但未与队列绑定，回调returnback接口
     */
    @Override
    public void sendTestMessage1(MqData mqData) {
        CorrelationData corrData = new CorrelationData(UUID.randomUUID().toString());
        logger.info("corrData：{}", corrData.getId());
        rabbitTemplate.convertAndSend(CommonConstans.DIRECT_EXCHANGE_NAME, "11", mqData.getMsg(), corrData);
    }
}
